package com.gaff.emp.core.hbase;

import com.gaff.emp.core.EmpException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class AbstractHBaseTemplate {

    private ThreadPoolExecutor pool = new ThreadPoolExecutor(20, 1000, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

    private Configuration configuration;

    public AbstractHBaseTemplate(Configuration configuration) {
        this.configuration = configuration;
        if (this.configuration == null) {
            throw new EmpException("a valid configuration is provided.");
        }
    }

    public AbstractHBaseTemplate(String quorum) {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set(HConstants.ZOOKEEPER_QUORUM, quorum);
        configuration.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 20 * 1000);
        configuration.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 5);
        configuration.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 30 * 1000);
        configuration.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 200000);
        this.configuration = configuration;
    }

    public Connection getConnection() throws IOException {
        return ConnectionFactory.createConnection(configuration, pool);
    }

    public Configuration getConfiguration() {
        return configuration;
    }

    public void close() {
        pool.shutdown();
    }

}
